package com.atguigu.gmall.realtime.dwd.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.util.JDBCUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;


public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>{
    private Map<String, TableProcessDim> configMap = new HashMap<>();
    private MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection mysqlConnection = JDBCUtil.getMysqlConnection();
        String sql = "select * from gmall0221_config.table_process_dim";
        JDBCUtil.queryList(mysqlConnection, sql, TableProcessDim.class, true);
        JDBCUtil.closeMysqlConnection(mysqlConnection);
    }
    // 10.1 processElement：处理主流上的业务数据 根据表名到广播状态中获取对应的配置，如果获取到了，说明是维度，将维度数据传递到下游
    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        String table = jsonObject.getString("table");
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        TableProcessDim tableProcessDim = null;

        if ((tableProcessDim = broadcastState.get(table)) != null
                || (tableProcessDim = configMap.get(table)) != null){
            JSONObject dataJsonObject = jsonObject.getJSONObject("data");
            String sinkColumns = tableProcessDim.getSinkColumns();
            deleteNotNeedColumn(dataJsonObject, sinkColumns);
            String type = jsonObject.getString("type");
            dataJsonObject.put("type", type);
            collector.collect(Tuple2.of(dataJsonObject, tableProcessDim));
        }
    }

    // 10.2 processBroadcastElement： 处理广播流处理数据， 将配置信息放到广播状态中或者删除对应的配置信息 K：维度表名 V： 实体对象
    @Override
    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //获取对配置表进行的操作类型
        String op = tableProcessDim.getOp();
        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);
        String sourceTable = tableProcessDim.getSourceTable();
        if ("d".equals(op)){
            //说明从配置信息中删除了一条配置，将对应的配置信息从广播状态中也删除掉
            broadcastState.remove(sourceTable);
        }else {
            //说明对配置表做了添加、读取、更新操作，将最新配置信息放到广播状态中
            broadcastState.put(sourceTable, tableProcessDim);
        }
    }

    public static void deleteNotNeedColumn(JSONObject dataJsonObject, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entries = dataJsonObject.entrySet();
//        Iterator<Map.Entry<String, Object>> it = entries.iterator();
//        for(;it.hasNext();){
//            Map.Entry<String, Object> entry = it.next();
//            if (!columnList.contains(entry.getKey())){
//                it.remove();
//            }
//        }
        entries.removeIf(entry -> !columnList.contains(entry.getKey()));
    }
}

